package com.szty.libarytest.longConnect.manager;

import android.content.Context;
import android.util.Log;

import com.szty.libarytest.longConnect.ConnectUtils;
import com.szty.libarytest.longConnect.client.FrameCodecFactory;
import com.szty.libarytest.longConnect.client.HeartBeatHandler;
import com.szty.libarytest.longConnect.client.HeartBeatListener;
import com.szty.libarytest.longConnect.client.HeartBeatMessageFactory;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.keepalive.KeepAliveFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import java.net.InetSocketAddress;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

import static com.szty.libarytest.longConnect.ConnectUtils.NOW_CONNECT_NUM;
import static com.szty.libarytest.longConnect.ConnectUtils.REPEAT_TIME;


public class ClientConnectManager {

    private static ClientConnectManager instance;
    private NioSocketConnector mSocketConnector;
    public Boolean isRePeatConnect = true;
    public static ClientConnectManager getInstance() {
        if (null == instance) {
            instance = new ClientConnectManager();
        }
        return instance;
    }

    private ClientConnectManager() {

    }

    private Context context;

    public void init(Context context) {
        this.context = context;
    }

    public void connect(final Context context) {
        isRePeatConnect = true;
        Observable.create(e -> {
            try{

            mSocketConnector = new NioSocketConnector();
            //设置协议封装解析处理
            mSocketConnector.getFilterChain().addLast("protocol", new ProtocolCodecFilter(new FrameCodecFactory()));
            //设置心跳包
            KeepAliveFilter heartFilter = new KeepAliveFilter(new HeartBeatMessageFactory());
            //每 10 秒钟发送一个心跳包
            heartFilter.setRequestInterval(ConnectUtils.SEND_HEART_TIME);
            //心跳包超时时间 10s
            heartFilter.setRequestTimeout(ConnectUtils.HEART_TIMEOUT_TIME);
            // 获取过滤器链
            DefaultIoFilterChainBuilder filterChain = mSocketConnector.getFilterChain();
            // 添加编码过滤器 处理乱码、编码问题
            filterChain.addLast("decoder", new ProtocolCodecFilter(new FrameCodecFactory()));
            mSocketConnector.getFilterChain().addLast("heartbeat", heartFilter);
            //设置 handler 处理业务逻辑
            mSocketConnector.setHandler(new HeartBeatHandler(context));
            mSocketConnector.addListener(new HeartBeatListener(mSocketConnector));
            //配置服务器地址
            InetSocketAddress mSocketAddress = new InetSocketAddress(ConnectUtils.HOST, ConnectUtils.PORT);
            //发起连接
            ConnectFuture mFuture = mSocketConnector.connect(mSocketAddress);

            mFuture.awaitUninterruptibly();
            IoSession mSession = mFuture.getSession();
            Log.d("", "======连接成功" + mSession.toString());
            e.onNext(mSession);
            e.onComplete();

            }catch ( Exception e1){
                e1.printStackTrace();
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Object o) {
                IoSession mSession = (IoSession) o;
                Log.d("MainActivity", "======连接成功了====" + mSession.isConnected());
                SessionManager.getInstance().setSeesion(mSession);
                //发消息给服务器，验证自身
//                SessionManager.getInstance().writeToServer(ConnectUtils.CHECKSELF);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }


    public void rePeatConnect() {
        NOW_CONNECT_NUM++;
        if (NOW_CONNECT_NUM >= REPEAT_TIME) {
            return;
        }
        final boolean[] isRepeat = {false};
        Observable.create(e -> {
            // 执行到这里表示Session会话关闭了，需要进行重连,我们设置每隔3s重连一次,如果尝试重连5次都没成功的话,就认为服务器端出现问题,不再进行重连操作
            int count = 0;// 记录尝试重连的次数
            while (!isRepeat[0] && count < 5) {
                try {
                    count++;
                        mSocketConnector = new NioSocketConnector();
                    //设置协议封装解析处理
                    mSocketConnector.getFilterChain().addLast("protocol", new ProtocolCodecFilter(new FrameCodecFactory()));
                    //设置心跳包
                    KeepAliveFilter heartFilter = new KeepAliveFilter(new HeartBeatMessageFactory());
                    //每 5 秒钟发送一个心跳包
                    heartFilter.setRequestInterval(ConnectUtils.SEND_HEART_TIME);
                    //心跳包超时时间 10s
                    heartFilter.setRequestTimeout(ConnectUtils.HEART_TIMEOUT_TIME);
                    // 获取过滤器链
                    DefaultIoFilterChainBuilder filterChain = mSocketConnector.getFilterChain();
                    filterChain.addLast("encoder", new ProtocolCodecFilter(new FrameCodecFactory()));
                    // 添加编码过滤器 处理乱码、编码问题
                    filterChain.addLast("decoder", new ProtocolCodecFilter(new FrameCodecFactory()));
                    mSocketConnector.getFilterChain().addLast("heartbeat", heartFilter);
                    //设置 handler 处理业务逻辑
                    mSocketConnector.setHandler(new HeartBeatHandler(context));
                    mSocketConnector.addListener(new HeartBeatListener(mSocketConnector));
                    //配置服务器地址
                    InetSocketAddress mSocketAddress = new InetSocketAddress(ConnectUtils.HOST, ConnectUtils.PORT);
                    //发起连接
                    ConnectFuture mFuture = mSocketConnector.connect(mSocketAddress);
                    mFuture.awaitUninterruptibly();
                    IoSession mSession = mFuture.getSession();
                    if (mSession.isConnected()) {
                        isRepeat[0] = true;
                        Log.d("", "======重连成功" + mSession.toString());
                        e.onNext(mSession);
                        e.onComplete();
                        break;
                    }
                } catch (Exception e1) {
                    if (count == REPEAT_TIME) {
                        System.out.println(ConnectUtils.stringNowTime() + " : 断线重连"
                                + REPEAT_TIME + "次之后仍然未成功,结束重连.....");
                        break;
                    } else {
                        System.out.println(ConnectUtils.stringNowTime() + " : 本次断线重连失败,3s后进行第" + (count + 1) + "次重连.....");
                        try {
                            Thread.sleep(3000);
                            System.out.println(ConnectUtils.stringNowTime() + " : 开始第" + (count + 1) + "次重连.....");
                        } catch (InterruptedException e12) {
                            Log.e("rePeatConnect ", "rePeatConnect e12" + e12);
                        }
                    }
                }

            }

        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull Object o) {
                IoSession mSession = (IoSession) o;
                Log.d("MainActivity", "======连接成功了吗====" + mSession.isConnected());
                NOW_CONNECT_NUM = 0;
                SessionManager.getInstance().setSeesion(mSession);
                SessionManager.getInstance().writeToServer("重新连接发起的请求");
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }
    public void disConnect(){
        isRePeatConnect = false;
        if(mSocketConnector == null ||mSocketConnector.isDisposed())return;
        mSocketConnector.dispose();
    }
    public void checkSocketStatus(){
        Log.e("SunJH",mSocketConnector.isDisposed()+"");
    }
}
